"""
Case Type   : gs_checkse工具功能
Case Name   : 验证gs_checkse的A5检测项确保撤销普通用户非必须的管理权限检测与修复成功
Create At   : 2024/10/28
Owner       : lihongji
Description :
    1.创建拥有CREATEROLE权限的普通用户后检测A5项
    2.修复A5项后再次检测并在数据库中查看CREATEROLE权限
    3.创建拥有CREATEDB权限的普通用户后检测A5项
    4.修复A5项后再次检测并在数据库中查看CREATEDB权限
    5.创建拥有AUDITADMIN权限的普通用户后检测A5项
    6.修复A5项后再次检测并在数据库中查看AUDITADMIN权限
    7.创建拥有MONADMIN权限的普通用户后检测A5项
    8.修复A5项后再次检测并在数据库中查看MONADMIN权限
    9.创建拥有OPRADMIN权限的普通用户后检测A5项
    10.修复A5项后再次检测并在数据库中查看OPRADMIN权限
    11.创建拥有POLADMIN权限的普通用户后检测A5项
    12.修复A5项后再次检测并在数据库中查看POLADMIN权限
Expect      :
    1.赋权成功，打印告警信息
    2.修复成功，不打印告警信息，查看CREATEROLE权限已撤销
    3.赋权成功，打印告警信息
    4.修复成功，不打印告警信息，查看CREATEDB权限已撤销
    5.赋权成功，打印告警信息
    6.修复成功，不打印告警信息，查看AUDITADMIN权限已撤销
    7.赋权成功，打印告警信息
    8.修复成功，不打印告警信息，查看MONADMIN权限已撤销
    9.赋权成功，打印告警信息
    10.修复成功，不打印告警信息，查看OPRADMIN权限已撤销
    11.赋权成功，打印告警信息
    12.修复成功，不打印告警信息，查看POLADMIN权限已撤销
History     :
"""
import os
import time
import unittest

from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from yat.test import Node
from yat.test import macro


class GS_CHECKSE(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'----{os.path.basename(__file__)} start----')
        self.primaryNode = Node('PrimaryDbUser')
        self.sh_primary = CommonSH('PrimaryDbUser')
        self.primaryDbA = Node('PrimaryDbAmode')
        self.primaryDbB = Node('PrimaryDbBmode')
        self.primaryDbC = Node('PrimaryDbCmode')
        self.primaryDbPG = Node('PrimaryDbPGmode')
        self.constant = Constant()
        self.common = Common()
        self.env_path = macro.DB_ENV_PATH

    def test_gscheckse(self):
        step = ('----step1:创建拥有CREATEROLE权限的普通用户后检测A5项 '
                'expect:赋权成功，打印告警信息----')
        self.log.info(step)
        createUser = ("drop user if exists lily;"
                      "CREATE user lily WITH CREATEROLE PASSWORD 'qwer*963.';")
        sql_res = self.sh_primary.execut_db_sql(createUser, dbname='postgres')
        gs_checkse_cmd = (f'source {self.env_path};'
                          f'gs_checkse -i A5 --detail')
        self.log.info(gs_checkse_cmd)
        check_res = self.common.get_sh_result(self.primaryNode, gs_checkse_cmd)
        self.assertIn("Ensure revocation of unnecessary administrative "
                      "privileges from regular users.",
                      check_res, '执行失败' + step)
        self.log.info(sql_res)
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG,
                      sql_res, '执行失败' + step)

        step = ('----step2:修复A5项后再次检测并在数据库中查看CREATEROLE权限  '
                'expect:修复成功，不打印告警信息，查看CREATEROLE权限已撤销----')
        self.log.info(step)
        gs_checkse_cmd = (f'source {self.env_path};'
                          f'gs_checkse -i B5 --detail;'
                          f'gs_checkse -i A5 --detail')
        self.log.info(gs_checkse_cmd)
        check_res = self.common.get_sh_result(self.primaryNode, gs_checkse_cmd)
        check_sql = ("SELECT rolname FROM pg_roles WHERE rolcreaterole = true"
                     " AND rolsuper = false;")
        result = self.sh_primary.execut_db_sql(check_sql, dbname='postgres')
        self.assertIn("Setting Permission management succeed",
                      check_res, '执行失败' + step)
        self.assertNotIn("Ensure revocation of unnecessary administrative "
                         "privileges from regular users.",
                         check_res, '执行失败' + step)
        self.assertIn('(0 rows)', result, '执行失败' + step)

        step = ('----step3:创建拥有CREATEDB权限的普通用户后检测A5项 '
                'expect:赋权成功，打印告警信息----')
        self.log.info(step)
        createUser = ("drop user if exists lily;"
                      "CREATE user lily WITH CREATEDB PASSWORD 'qwer*963.';")
        sql_res = self.sh_primary.execut_db_sql(createUser, dbname='postgres')
        gs_checkse_cmd = (f'source {self.env_path};'
                          f'gs_checkse -i A5 --detail')
        self.log.info(gs_checkse_cmd)
        check_res = self.common.get_sh_result(self.primaryNode, gs_checkse_cmd)
        self.assertIn("Ensure revocation of unnecessary administrative "
                      "privileges from regular users.",
                      check_res, '执行失败' + step)
        self.log.info(sql_res)
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG,
                      sql_res, '执行失败' + step)

        step = ('----step4:修复A5项后再次检测并在数据库中查看CREATEDB权限  '
                'expect:修复成功，不打印告警信息，查看CREATEDB权限已撤销----')
        self.log.info(step)
        gs_checkse_cmd = (f'source {self.env_path};'
                          f'gs_checkse -i B5 --detail;'
                          f'gs_checkse -i A5 --detail')
        self.log.info(gs_checkse_cmd)
        check_res = self.common.get_sh_result(self.primaryNode, gs_checkse_cmd)
        check_sql = ("SELECT rolname FROM pg_roles WHERE rolcreatedb = true "
                     "AND rolsuper = false;")
        result = self.sh_primary.execut_db_sql(check_sql, dbname='postgres')
        self.assertIn("Setting Permission management succeed",
                      check_res, '执行失败' + step)
        self.assertNotIn("Ensure revocation of unnecessary administrative "
                         "privileges from regular users.",
                         check_res, '执行失败' + step)
        self.assertIn('(0 rows)', result, '执行失败' + step)

        step = ('----step5:创建拥有AUDITADMIN权限的普通用户后检测A5项 '
                'expect:赋权成功，打印告警信息----')
        self.log.info(step)
        createUser = ("drop user if exists lily;"
                      "CREATE user lily WITH AUDITADMIN PASSWORD 'qwer*963.';")
        sql_res = self.sh_primary.execut_db_sql(createUser, dbname='postgres')
        gs_checkse_cmd = (f'source {self.env_path};'
                          f'gs_checkse -i A5 --detail')
        self.log.info(gs_checkse_cmd)
        check_res = self.common.get_sh_result(self.primaryNode, gs_checkse_cmd)
        self.assertIn("Ensure revocation of unnecessary administrative "
                      "privileges from regular users.",
                      check_res, '执行失败' + step)
        self.log.info(sql_res)
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG,
                      sql_res, '执行失败' + step)

        step = ('----step6:修复A5项后再次检测并在数据库中查看AUDITADMIN权限  '
                'expect:修复成功，不打印告警信息，查看AUDITADMIN权限已撤销----')
        self.log.info(step)
        gs_checkse_cmd = (f'source {self.env_path};'
                          f'gs_checkse -i B5 --detail;'
                          f'gs_checkse -i A5 --detail')
        self.log.info(gs_checkse_cmd)
        check_res = self.common.get_sh_result(self.primaryNode, gs_checkse_cmd)
        check_sql = ("SELECT rolname FROM pg_roles WHERE rolauditadmin = "
                     "true AND rolsuper = false;")
        result = self.sh_primary.execut_db_sql(check_sql, dbname='postgres')
        self.assertIn("Setting Permission management succeed",
                      check_res, '执行失败' + step)
        self.assertNotIn("Ensure revocation of unnecessary administrative "
                         "privileges from regular users.",
                         check_res, '执行失败' + step)
        self.assertIn('(0 rows)', result, '执行失败' + step)

        step = ('----step7:创建拥有MONADMIN权限的普通用户后检测A5项 '
                'expect:赋权成功，打印告警信息----')
        self.log.info(step)
        createUser = ("drop user if exists lily;"
                      "CREATE user lily WITH MONADMIN PASSWORD 'qwer*963.';")
        sql_res = self.sh_primary.execut_db_sql(createUser, dbname='postgres')
        gs_checkse_cmd = (f'source {self.env_path};'
                          f'gs_checkse -i A5 --detail')
        self.log.info(gs_checkse_cmd)
        check_res = self.common.get_sh_result(self.primaryNode, gs_checkse_cmd)
        self.assertIn("Ensure revocation of unnecessary administrative "
                      "privileges from regular users.",
                      check_res, '执行失败' + step)
        self.log.info(sql_res)
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG,
                      sql_res, '执行失败' + step)

        step = ('----step8:修复A5项后再次检测并在数据库中查看MONADMIN权限  '
                'expect:修复成功，不打印告警信息，查看MONADMIN权限已撤销----')
        self.log.info(step)
        gs_checkse_cmd = (f'source {self.env_path};'
                          f'gs_checkse -i B5 --detail;'
                          f'gs_checkse -i A5 --detail')
        self.log.info(gs_checkse_cmd)
        check_res = self.common.get_sh_result(self.primaryNode, gs_checkse_cmd)
        check_sql = ("SELECT rolname FROM pg_roles WHERE "
                     "rolmonitoradmin = true AND rolsuper = false;")
        result = self.sh_primary.execut_db_sql(check_sql, dbname='postgres')
        self.assertIn("Setting Permission management succeed",
                      check_res, '执行失败' + step)
        self.assertNotIn("Ensure revocation of unnecessary administrative "
                         "privileges from regular users.",
                         check_res, '执行失败' + step)
        self.assertIn('(0 rows)', result, '执行失败' + step)

        step = ('----step9:创建拥有OPRADMIN权限的普通用户后检测A5项 '
                'expect:赋权成功，打印告警信息----')
        self.log.info(step)
        createUser = ("drop user if exists lily;"
                      "CREATE user lily WITH OPRADMIN PASSWORD 'qwer*963.';")
        sql_res = self.sh_primary.execut_db_sql(createUser, dbname='postgres')
        gs_checkse_cmd = (f'source {self.env_path};'
                          f'gs_checkse -i A5 --detail')
        self.log.info(gs_checkse_cmd)
        check_res = self.common.get_sh_result(self.primaryNode, gs_checkse_cmd)
        self.assertIn("Ensure revocation of unnecessary administrative "
                      "privileges from regular users.",
                      check_res, '执行失败' + step)
        self.log.info(sql_res)
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG,
                      sql_res, '执行失败' + step)

        step = ('----step10:修复A5项后再次检测并在数据库中查看OPRADMIN权限  '
                'expect:修复成功，不打印告警信息，查看OPRADMIN权限已撤销----')
        self.log.info(step)
        gs_checkse_cmd = (f'source {self.env_path};'
                          f'gs_checkse -i B5 --detail;'
                          f'gs_checkse -i A5 --detail')
        self.log.info(gs_checkse_cmd)
        check_res = self.common.get_sh_result(self.primaryNode, gs_checkse_cmd)
        check_sql = ("SELECT rolname FROM pg_roles WHERE "
                     "roloperatoradmin = true AND rolsuper = false;")
        result = self.sh_primary.execut_db_sql(check_sql, dbname='postgres')
        self.assertIn("Setting Permission management succeed",
                      check_res, '执行失败' + step)
        self.assertNotIn("Ensure revocation of unnecessary administrative "
                         "privileges from regular users.",
                         check_res, '执行失败' + step)
        self.assertIn('(0 rows)', result, '执行失败' + step)

        step = ('----step11:创建拥有POLADMIN权限的普通用户后检测A5项 '
                'expect:赋权成功，打印告警信息----')
        self.log.info(step)
        createUser = ("drop user if exists lily;"
                      "CREATE user lily WITH POLADMIN PASSWORD 'qwer*963.';")
        sql_res = self.sh_primary.execut_db_sql(createUser, dbname='postgres')
        gs_checkse_cmd = (f'source {self.env_path};'
                          f'gs_checkse -i A5 --detail')
        self.log.info(gs_checkse_cmd)
        check_res = self.common.get_sh_result(self.primaryNode, gs_checkse_cmd)
        self.assertIn("Ensure revocation of unnecessary administrative "
                      "privileges from regular users.",
                      check_res, '执行失败' + step)
        self.log.info(sql_res)
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG,
                      sql_res, '执行失败' + step)

        step = ('----step12:修复A5项后再次检测并在数据库中查看POLADMIN权限  '
                'expect:修复成功，不打印告警信息，查看POLADMIN权限已撤销----')
        self.log.info(step)
        gs_checkse_cmd = (f'source {self.env_path};'
                          f'gs_checkse -i B5 --detail;'
                          f'gs_checkse -i A5 --detail')
        self.log.info(gs_checkse_cmd)
        check_res = self.common.get_sh_result(self.primaryNode, gs_checkse_cmd)
        check_sql = ("SELECT rolname FROM pg_roles WHERE rolpolicyadmin = true"
                     " AND rolsuper = false;")
        result = self.sh_primary.execut_db_sql(check_sql, dbname='postgres')
        self.assertIn("Setting Permission management succeed",
                      check_res, '执行失败' + step)
        self.assertNotIn("Ensure revocation of unnecessary administrative "
                         "privileges from regular users.",
                         check_res, '执行失败' + step)
        self.assertIn('(0 rows)', result, '执行失败' + step)

    def tearDown(self):
        step = '----step13:清理环境 expect:成功----'
        self.log.info(step)
        revoke_sql = (
            f"drop user if exists lily;"
            f"grant create on schema public to {self.primaryNode.ssh_user};"
            f"grant all privileges on database {self.primaryNode.db_name} "
            f"to {self.primaryNode.ssh_user};"
            f"grant all privileges on database {self.primaryDbA.db_name} "
            f"to {self.primaryDbA.ssh_user};;"
            f"grant all privileges on database {self.primaryDbB.db_name} "
            f"to {self.primaryDbB.ssh_user};;"
            f"grant all privileges on database {self.primaryDbC.db_name} "
            f"to {self.primaryDbC.ssh_user};;"
            f"grant all privileges on database {self.primaryDbPG.db_name} "
            f"to {self.primaryDbPG.ssh_user};"
            f"")
        sql_res = self.sh_primary.execut_db_sql(revoke_sql, dbname='postgres')
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG, sql_res,
                      '执行失败' + step)
        self.log.info(f'----{os.path.basename(__file__)} end----')